package bsf.baseservice.businessmq;

/**
 * Created by chejiangyi on 2017/4/20.
 */

import bsf.base.CallBack;
import bsf.baseservice.businessmq.common.BusinessMQConfig;
import bsf.baseservice.businessmq.common.BusinessMQResponse;
import bsf.baseservice.businessmq.consumer.ConsumerProvider;
import bsf.baseservice.businessmq.model.tb_mqpath_partition_model;
import bsf.baseservice.businessmq.producter.ProducterPoolHelper;
import bsf.baseservice.businessmq.producter.ProducterProvider;
import bsf.baseservice.businessmq.producter.loadbalance.BaseLoadBalance;
import bsf.system.thread.ThreadPool;
import bsf.util.ExceptionUtil;

/**
 * BusinessMQ sdk Demo
 */
public class BusinessMQSdkDemo {
    public static void main(String[] args)
    {

        ThreadPool.SystemThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    BusinessMQSdkDemo sdkDemo = new BusinessMQSdkDemo();
                    for (int i = 0; i < 1000; i++) {
                        sdkDemo.sendMessageDemo("发送消息" + i);
                        System.err.println("【发送消息】"+i);
                        try {
                            Thread.sleep(1000 * 2);
                        } catch (Exception e) {
                        }
                    }
                }
                catch (Throwable exp)
                {
                    System.err.println(ExceptionUtil.getFullMessage(exp));
                }
            }
        });
      ThreadPool.SystemThreadPool.execute(new Runnable() {
          @Override
          public void run() {
              try {
                  BusinessMQSdkDemo sdkDemo = new BusinessMQSdkDemo();
                  sdkDemo.receiveMessageDemo(new CallBack.Action1<String>() {
                      @Override
                      public void invoke(String s) {
                          System.err.println("【收到消息】" + s);
                      }
                  });
              }catch (Exception exp)
              {
                  System.err.println(ExceptionUtil.getFullMessage(exp));
              }
          }
      });

       try{ System.in.read();}catch (Exception exp){}
    }

    public final void sendMessageDemo(String msg) {
        //发送字符串示例
        BusinessMQConfig t = new BusinessMQConfig();
        t.ManageConnectString = "type=sqlserver;pooltype=druid;server=192.168.1.209:1433;Database=dyd_bs_MQ_manage;username=sa;password=111;";
        ProducterProvider p = ProducterPoolHelper.getPool(t, "ddd"); //队列路径.分隔,类似类的namespace,是队列的唯一标识，要提前告知运维在消息中心注册，方可使用。 - 管理中心数据库 - （如果采用系统自带的算法，则不需要重设属性，一般建议采用系统自带的负载均衡算法）设置自定义的负载均衡算法，默认使用系统自带的SystemLoadBalance算法。
        p.sendMessage(msg);
        //发送对象示例
//             var obj = new message2 { text = "文字", num = 1 };
//              var p = ProducterPoolHelper.GetPool(new BusinessMQConfig() { ManageConnectString = "server=192.168.17.237;Initial Catalog=dyd_bs_MQ_manage;User ID=sa;Password=Xx~!@#;" },//管理中心数据库
//                "test.diayadian.obj");//队列路径 .分隔,类似类的namespace,是队列的唯一标识，要提前告知运维在消息中心注册，方可使用。
//            p.SendMessage<message>(obj);
//
    }

    private ConsumerProvider Consumer;

    public final void receiveMessageDemo(final CallBack.Action1<String> action) {
        if (Consumer == null) {
            Consumer = new ConsumerProvider();
            Consumer.setClient("ddd.customer1"); //clientid,接收消息的(消费者)唯一标示,一旦注册以后，不能更改，业务下线废弃后必须要告知运维，删除消费者注册。
            Consumer.ClientName = "客户端名称"; //这个相对随意些，主要是用来自己识别的，要简短
            BusinessMQConfig t = new BusinessMQConfig();
            t.ManageConnectString = "server=192.168.1.209;Initial Catalog=dyd_bs_MQ_manage;User ID=sa;Password=111;";
            Consumer.Config = t;
            Consumer.MaxReceiveMQThread = 1; //并行处理的线程数,一般为1足够,若消息处理慢,又想并行消费,则考虑 正在使用的分区=并行处理线程数 为并行效率极端最优,但cpu消耗应该不小。
            Consumer.setMQPath("ddd"); //接收的队列要正确
            Consumer.PartitionIndexs = new java.util.ArrayList<Integer>(java.util.Arrays.asList(new Integer[]{1, 2, 3, 4, 5, 6,  8,7,7})); //消费者订阅的分区顺序号,从1开始

//                       * 这些编写业务代码
//                       * 编写的时候要注意考虑，业务处理失败的情况。
//                       * 1.重试失败n次。
//                       * 2.重试还不行，则标记消息已被处理。然后跳过该消息处理，自己另外文档记录这种情况。
//                       * 消息被消费完毕，一定要调用MarkFinished，标记消息被消费完毕。
//
            Consumer.registerReceiveMQListener(new CallBack.Action1<BusinessMQResponse<String>>() {
                                                   @Override
                                                   public void invoke(BusinessMQResponse<String> r) {
                                                       action.invoke(r.ObjMsg);
                                                       r.markFinished();
                                                   }
                                               }
                    , String.class);


        }

    }

    /**
     * 关闭消息订阅连接
     */
    public final void closeReceiveMessage() {
        //注册消费者消息,消费者务必要在程序关闭后关掉（dispose）。否则导致异常终止,要人工等待连接超时后，方可重新注册。
        if (Consumer != null) {
            Consumer.close();
            Consumer = null;
        }
    }
    public static   class message2 {
        public String text;
        public int num;

    }
}

